package com.educate.device.message;

import com.educate.common.web.JsonUtils;
import com.educate.device.dao.DeviceMessageDao;
import com.educate.device.model.po.DeviceMessage;
import com.google.common.base.Charsets;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Timestamp;

@Component
@Log4j2
public class MessageServer implements InitializingBean {

    @Resource DeviceMessageDao deviceMessageDao;


    public static void main(String[] args) throws Exception {
        boolean debugEnabled = log.isDebugEnabled();
        log.debug("debugEnabled=" + debugEnabled);
        new MessageServer().afterPropertiesSet();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        String serverURI = "tcp://114.215.69.197:1883";
        String clientId = "service-device";
        MqttClient mqttClient = new MqttClient(serverURI, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
//        options.setUserName();
//        options.setPassword();
//        options.setConnectionTimeout(10);
//        options.setKeepAliveInterval(60);
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.debug("connectionLost");
                cause.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                byte[] payload = message.getPayload();
                String content = new String(payload, Charsets.UTF_8);
                DeviceMessage deviceMessage = JsonUtils.readValue(content, DeviceMessage.class);
                deviceMessage.setCreateTime(new Timestamp(System.currentTimeMillis()));
                if (DeviceMessage.TYPE_OPERATE.equals(deviceMessage.getType())) {
                    deviceMessageDao.save(deviceMessage);
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.debug("deliveryComplete");
            }
        });
        mqttClient.connect(options);
//        mqttClient.getTopic("/World");
        mqttClient.subscribe("/World");

        MqttMessage message = new MqttMessage();

        mqttClient.getTopic("aaa").publish(message);
    }
}
